package com.im.business.history;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @Created leijun
 * @Date 2021/10/14 10:37 下午
 */
public class KafkaManager {

    private KafkaManager() {}

    private static class Singleton{
        private static KafkaManager instance = new KafkaManager();
    }

    public static KafkaManager getInstance() {
        return Singleton.instance;
    }

    private static KafkaConsumer<String, String> consumer;

    private static KafkaProducer<String, String> producer;

    static {
        Properties configs = new Properties();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");//kafka地址，多个地址用逗号分割
        configs.put("group.id","business_group");
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configs.put("key.deserializer", StringDeserializer.class);
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configs.put("value.deserializer", StringDeserializer.class);
        consumer = new KafkaConsumer<String, String>(configs);

        producer = new KafkaProducer<String, String>(configs);
    }

    public KafkaConsumer getConsumer() {
        return consumer;
    }

    public static KafkaProducer<String, String> getProducer() {
        return producer;
    }
}
